package org.joy.hpc.analyzer.examples.sogou;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SkipBadRecords;
import org.joy.hpc.analyzer.HitMapper;
import org.joy.hpc.analyzer.HitReducer;
import org.joy.hpc.analyzer.HitWritable;
import org.joy.hpc.analyzer.IndexDBOutputFormat;

/**
 * 用搜狗语料作分析Hit关键词的driver
 * 
 * @author Song Liu (Lamfeeling@126.com)
 */
public class SogouHitAnalysisDriver {
	private static final Log LOG = LogFactory
			.getLog(SogouHitAnalysisDriver.class.getName());

	public static void runJob(String input, String output, String dictsHomeCache)
			throws IOException {

		JobClient client = new JobClient();
		JobConf conf = new JobConf(SogouHitAnalysisDriver.class);

		FileSystem fs = FileSystem.get(conf);
		Path outPath = new Path(output);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(HitWritable.class);
		FileInputFormat.setInputPaths(conf, new Path(input));
//		FileOutputFormat.setOutputPath(conf, outPath);

		conf.setMapperClass(HitMapper.class);
		conf.setReducerClass(HitReducer.class);
		conf.setInputFormat(SogouInputFormat.class);
		conf.setOutputFormat(IndexDBOutputFormat.class);
		
		SkipBadRecords.setMapperMaxSkipRecords(conf, 1);
		SkipBadRecords.setAttemptsToStartSkipping(conf, 1);
		SkipBadRecords.setSkipOutputPath(conf, null);

		if (dictsHomeCache != null) {
			DistributedCache.createSymlink(conf);
			try {
				DistributedCache.addCacheArchive(new URI(dictsHomeCache+"#dicts"), conf);
			} catch (URISyntaxException e1) {
				// if URI not specified
				e1.printStackTrace();
			}
		} else {
			LogFactory.getLog(SogouHitAnalysisDriver.class).warn(
					"Dicts Cache Not Specified!");
		}
		client.setConf(conf);

		try {
			JobClient.runJob(conf);
		} catch (IOException e) {
			LOG.error(e.toString());
		}
	}

	public static void main(String[] args) throws IOException {
		String input, output;
		if (args.length < 2) {
			input = "./input/";
			output = "./output/";
		} else {
			input = args[0];
			output = args[1];
		}
		SogouHitAnalysisDriver.runJob(input, output, args.length == 3 ? args[2]
				: null);
	}

}
